package cn.itcast.streaming.function.flatmap;

import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.bean.VehicleInfoModel;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;

//自定义实现告警规则窗口流与车辆基础信息广播流数据的拉宽操作
public class CustomRuleAlarmVehicleInfoFunction implements CoFlatMapFunction<ArrayList<ItcastDataPartObj>, HashMap<String, VehicleInfoModel>,ArrayList<Tuple2<ItcastDataPartObj,VehicleInfoModel>>> {
    private HashMap<String,VehicleInfoModel> infoMap = new HashMap<>();
    @Override
    public void flatMap1(ArrayList<ItcastDataPartObj> itcastDataPartObjs, Collector<ArrayList<Tuple2<ItcastDataPartObj, VehicleInfoModel>>> collector) throws Exception {
        ArrayList<Tuple2<ItcastDataPartObj, VehicleInfoModel>> tupleList = new ArrayList<>();

        for (ItcastDataPartObj obj : itcastDataPartObjs) {
            Tuple2<ItcastDataPartObj, VehicleInfoModel> tuple2 = new Tuple2<>();
            String vin = obj.getVin();
            VehicleInfoModel vehicleInfoModel = infoMap.getOrDefault(vin, null);
            tuple2.f0 = obj;
            tuple2.f1 = vehicleInfoModel;
            tupleList.add(tuple2);
        }
        collector.collect(tupleList);
    }

    @Override
    public void flatMap2(HashMap<String, VehicleInfoModel> stringVehicleInfoModelHashMap, Collector<ArrayList<Tuple2<ItcastDataPartObj, VehicleInfoModel>>> collector) throws Exception {
        infoMap = stringVehicleInfoModelHashMap;
    }
}
